#include "threadpool_win.h"

struct ThreadPool * threadpool_init(int thread_max_num, int queue_max_num)
{
	struct ThreadPool * pool = NULL;
	do {
		pool = (struct ThreadPool *)malloc(sizeof(struct ThreadPool));
		if (pool == NULL) {
			break;
		}
		pool->thread_num = thread_max_num;
		pool->queue_max_num = queue_max_num;
		pool->queue_cur_num = 0;
		pool->head = NULL;
		pool->tail = NULL;
		
		InitializeConditionVariable(&pool->cv_empty);
		InitializeConditionVariable(&pool->cv_not_empty);
		InitializeConditionVariable(&pool->cv_not_full);
		InitializeCriticalSection(&pool->cs);

		pool->hThreads = (PHANDLE)malloc(sizeof(HANDLE) * pool->thread_num);
		if (pool->hThreads == NULL) {
			free(pool);
			pool = NULL;
			break;
		}

		pool->queue_close = 0;
		pool->pool_close = 0;
		for (int i = 0; i < pool->thread_num; i++) {
			unsigned int uiThreadId = 0;
			pool->hThreads[i] = (HANDLE)_beginthreadex(NULL, 0, threadpool_func, (void *)pool, 0, &uiThreadId);
		}

	} while (0);
	return pool;
}

int threadpool_add_job(struct ThreadPool * pool_, unsigned int(__stdcall *callback_func_)(void *), void * param_)
{
	if (pool_ == NULL || callback_func_ == NULL) {
		return -1;
	}
	EnterCriticalSection(&pool_->cs);

	while (pool_->queue_cur_num == pool_->queue_max_num && !(pool_->queue_close || pool_->pool_close)) {
		SleepConditionVariableCS(&pool_->cv_not_full, &pool_->cs, INFINITE);
	}

	if (pool_->queue_close || pool_->pool_close) {
		LeaveCriticalSection(&pool_->cs);
		return -1;
	}

	struct Job * pJob = (struct Job *)malloc(sizeof(struct Job));
	if (pJob == NULL) {
		LeaveCriticalSection(&pool_->cs);
		return -1;
	}

	pJob->callback_func = callback_func_;
	pJob->param = param_;
	pJob->next = NULL;
	if (pool_->head == NULL) {
		pool_->head = pool_->tail = pJob;
		WakeConditionVariable(&(pool_->cv_not_empty));
	}
	else {
		pool_->tail->next = pJob;
		pool_->tail = pJob;
	}
	pool_->queue_cur_num++;
	LeaveCriticalSection(&pool_->cs);
	return 0;
}

unsigned int __stdcall threadpool_func(void * param_)
{
	struct ThreadPool * pool = (struct ThreadPool *)param_;
	struct Job * pJob = NULL;
	while (1) {
		EnterCriticalSection(&pool->cs);
		while (pool->queue_cur_num == 0 && !pool->pool_close) {
			SleepConditionVariableCS(&pool->cv_not_empty, &pool->cs, INFINITE);
		}
		if (pool->pool_close) {
			LeaveCriticalSection(&pool->cs);
			break;
		}
		pool->queue_cur_num--;
		pJob = pool->head;
		if (pool->queue_cur_num == 0) {
			pool->head = pool->tail = NULL;
		}
		else {
			pool->head = pJob->next;
		}
		if (pool->queue_cur_num == 0) {
			WakeConditionVariable(&pool->cv_empty);
		}
		if (pool->queue_cur_num < pool->queue_max_num - 1) {
			WakeConditionVariable(&pool->cv_not_full);
		}
		LeaveCriticalSection(&pool->cs);
		pJob->callback_func(pJob->param);
		free(pJob);
		pJob = NULL;
	}
	return 0;
}

int threadpool_destroy(struct ThreadPool * pool_)
{
	if (pool_) {
		EnterCriticalSection(&pool_->cs);
		if (pool_->queue_close || pool_->pool_close) {
			LeaveCriticalSection(&pool_->cs);
			return -1;
		}
		pool_->queue_close = 1;
		while (pool_->queue_cur_num > 0) {
			SleepConditionVariableCS(&pool_->cv_empty, &pool_->cs, INFINITE);
		}
		pool_->pool_close = 1;
		LeaveCriticalSection(&pool_->cs);
		WakeAllConditionVariable(&pool_->cv_not_empty);
		WakeAllConditionVariable(&pool_->cv_not_full);
		WaitForMultipleObjects(pool_->thread_num, pool_->hThreads, TRUE, INFINITE);
		for (int i = 0; i < pool_->thread_num; i++) {
			CloseHandle(pool_->hThreads[i]);
			pool_->hThreads[i] = NULL;
		}
		free(pool_->hThreads);
		pool_->hThreads	= NULL;
		struct Job * p;
		while (pool_->head != NULL) {
			p = pool_->head;
			pool_->head = p->next;
			free(p);
		}
		free(pool_);
		pool_ = NULL;
	}
	return 0;
}